package com.bkhech.netty.httpwebsocket.client;

import cn.hutool.core.util.RandomUtil;
import com.bkhech.netty.httpwebsocket.pojo.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.util.concurrent.Promise;

import java.net.URI;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * WebSocketNettyClient
 * Java Netty websocket 客户端
 *
 * @author guowm
 * @date 2024-11-13
 */
public class WebSocketNettyClient {
    private final static ClientHandler clientHandler = new ClientHandler();

    private static volatile WebSocketNettyClient client;

    private WebSocketNettyClient() {
    }

    public static WebSocketNettyClient getWebSocketClient() {
        if (client == null) {
            synchronized (WebSocketNettyClient.class) {
                if (client == null) {
                    client = new WebSocketNettyClient();
                }
            }
        }
        return client;
    }

    private void connect(String wsServerUri, String clientName, Promise<Boolean> connectFuture) {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 因为 WebSocket 是基于http协议，使用http的编码和解码器
                            pipeline.addLast(new HttpClientCodec());
                            // 用于支持大数据流
                            pipeline.addLast(new ChunkedWriteHandler());

                            /*
                                1.http数据在传输过程中是分段，HttpObjectAggregator就是可以将多个段聚合，这就解释了当浏览器发送大量数据时，会发出多次http请求的问题
                                2.这个聚合器主要是将 HttpMessage 聚合成http为一个完整的报文，即 FullHttpRequest/FullHttpResponse
                             */
                            pipeline.addLast(new HttpObjectAggregator(65536));

                            pipeline.addLast(clientHandler);
                        }
                    });

            URI websocketURI = new URI(wsServerUri);
            HttpHeaders httpHeaders = new DefaultHttpHeaders();
            // 处理握手操作，！！！关键步骤！！！
            WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(websocketURI, WebSocketVersion.V13, null, true, httpHeaders);
            // 启动客户端
            final ChannelFuture channelFuture = bootstrap.connect(websocketURI.getHost(), websocketURI.getPort()).sync();
            final Channel channel = channelFuture.channel();
            // 发送握手请求
            handshaker.handshake(channel);
            clientHandler.setHandshaker(handshaker);

            //阻塞等待是否握手成功，直至 channelRead0() 方法中的  this.handshakeFuture.setSuccess(); 这个代码设置结果后返回
            final ChannelFuture handshakeFutureResult = clientHandler.handshakeFuture().sync();
            if (!handshakeFutureResult.isSuccess()) {
                startupFailure(clientName, null);
            }

            System.out.println("客户端 " + clientName + " 启动成功！");
            // 设置握手结果，让主线程继续执行
            connectFuture.setSuccess(true);

            // 阻塞等待连接被关闭
            channel.closeFuture().sync();
        } catch (Exception e) {
            startupFailure(clientName, e);
        } finally {
            group.shutdownGracefully();
            System.out.println("客户端 " + clientName + " 已关闭！");
        }
    }

    private static void startupFailure(String clientName, Exception e) {
        System.out.println("客户端 " + clientName + " 启动失败！");
        if (e != null) {
            e.printStackTrace();
        }
        // 连接失败，退出
        System.exit(-1);
    }

    /**
     * ClientHandler
     * 客户端业务处理类
     *
     * @author guowm
     * @date 2024-11-13
     */
    public static class ClientHandler extends SimpleChannelInboundHandler<Object> {

        private WebSocketClientHandshaker handshaker;
        private ChannelPromise handshakeFuture;
        private Channel channel;

        /**
         * 当客户端主动链接服务端的链接后，调用此方法
         *
         * @param ctx ChannelHandlerContext
         */
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) {
            System.out.println("客户端 handlerAdded .....");
            this.handshakeFuture = ctx.newPromise();
            this.channel = ctx.channel();
        }

        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            System.out.println("客户端 handlerRemoved .....");
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            System.out.println("客户端 exception: " + cause.getMessage());
            ctx.close();
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object o) throws Exception {
            // 握手协议返回，设置结束握手
            if (!this.handshaker.isHandshakeComplete()) {
                FullHttpResponse response = (FullHttpResponse) o;
                // 完成握手 WebSocketClientHandshaker，！！！关键步骤！！！
                this.handshaker.finishHandshake(ctx.channel(), response);
                // 设置握手状态为成功，
                // final ChannelFuture handshakeFuture = clientHandler.handshakeFuture().sync(); 这行代码阻塞结束
                this.handshakeFuture.setSuccess();
                // 接下来就可以正常通信了
                System.out.println("ClientHandler handshake: 握手成功!");
            } else if (o instanceof TextWebSocketFrame) {
                TextWebSocketFrame textFrame = (TextWebSocketFrame) o;
                System.out.println("服务器响应 textFrame: " + textFrame.text());
            } else if (o instanceof CloseWebSocketFrame) {
                // 服务器主动关闭连接时调用，ctx.close();
                System.out.println("服务器响应 CloseWebSocketFrame");
            } else {
                System.out.println("服务器响应 noop，未知消息类型");
            }
        }

        public void setHandshaker(WebSocketClientHandshaker handshaker) {
            this.handshaker = handshaker;
        }

        public ChannelFuture handshakeFuture() {
            return this.handshakeFuture;
        }

        public Channel getChannel() {
            return channel;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 登录凭证
        final String token = "123456";
        String serverUri = "ws://localhost:8080/hello";
        final String clientName = UUID.randomUUID().toString().substring(0, 8);

        Promise<Boolean> connectFuture = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
        // 异步启动 netty 客户端，防止阻塞
        Executors.newSingleThreadExecutor().execute(() -> getWebSocketClient().connect(serverUri, clientName, connectFuture));

        // 等待连接就绪，否则阻塞主线程
        final Promise<Boolean> connectResult = connectFuture.sync();
        if (!connectResult.isSuccess()) {
            System.out.println("客户端启动失败！");
        }

        // 接下来可以进行业务处理
        // 登录 LoginMessage， senderId 可以是客户端的唯一标识，比如 Android 设备的 androidId 或者 userId
        String senderId = "1";
        LoginMessage loginMessage = LoginMessage.builder().msgId(RandomUtil.getRandom().nextLong() + "").senderId(senderId).token(token).build();
        Request loginRequest = Request.builder().msgType(MsgType.CONNECT_INIT).message(loginMessage).build();
        clientHandler.getChannel().writeAndFlush(new TextWebSocketFrame(loginRequest.toJsonStr()));
        System.out.println("成功发送消息：" + loginRequest.toJsonStr());

        Thread.sleep(1000);

        // 心跳信息
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(() -> {
            Message keepAliveMessage = Message.builder().msgId(RandomUtil.getRandom().nextLong() + "").senderId(senderId).token(token).build();
            Request keepAliveRequest = Request.builder().msgType(MsgType.KEEPALIVE).message(keepAliveMessage).build();
            final ChannelFuture channelFuture = clientHandler.getChannel().writeAndFlush(new TextWebSocketFrame(keepAliveRequest.toJsonStr()));
            System.out.println("成功发送消息：" + keepAliveRequest.toJsonStr());
            try {
                channelFuture.get(2000, TimeUnit.MILLISECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                e.printStackTrace();
                clientHandler.getChannel().close();
                // TODO 此处可以写重连逻辑
            }
        }, 1500, 3000, TimeUnit.MILLISECONDS);

        Thread.sleep(1000);
        // 发送消息
        ChatMessage chatMessage = ChatMessage.builder()
                .msgId(RandomUtil.getRandom().nextLong() + "")
                .senderId(senderId).token(token).receiverId("2")
                .chatType(ChatType.WORDS).msg("你好").build();
        Request chatRequest = Request.builder().msgType(MsgType.CHAT).message(chatMessage).build();
        clientHandler.getChannel().writeAndFlush(new TextWebSocketFrame(chatRequest.toJsonStr()));
        System.out.println("成功发送消息：" + chatRequest.toJsonStr());
    }

}